package com.boot.study.table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors._

object MysqlOutputTest {
  def main(args: Array[String]): Unit = {
    // 1: 创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    val filePath: String = "D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt"
    tableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("inputTable")
    val sensorTable: Table = tableEnv.from("inputTable")
    // 聚合
    val aggTable: Table = sensorTable
      .groupBy('id) // 根据id分组
      .select('id, 'id.count as 'count)

    /**
     * 实际mysql的表
     * CREATE TABLE `sensor_count` (
     * `id` varchar(10) NOT NULL COMMENT 'id',
     * `cnt` bigint(5) NOT NULL COMMENT '次数',
     * PRIMARY KEY (`id`)
     * ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
     */
    // 输出es
    val sinkDDL =
    """
      |create table jdbcOutputTable(
      |  id varchar(10) not null,
      |  cnt bigint not null
      |) with (
      | 'connector.type'= 'jdbc',
      | 'connector.url'= 'jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&serverTimezone=UTC',
      | 'connector.table'= 'sensor_count',
      | 'connector.driver'= 'com.mysql.jdbc.Driver',
      | 'connector.username'= 'root',
      | 'connector.password'= '123'
      |)
      |""".stripMargin
    tableEnv.sqlUpdate(sinkDDL)
    aggTable.insertInto("jdbcOutputTable")

    env.execute("mysql output test")
  }
}
